package com.bwie.common.config;


import lombok.extern.log4j.Log4j2;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;

@Configuration
@Log4j2
public class MqConfig implements RabbitTemplate.ReturnsCallback,RabbitTemplate.ConfirmCallback {

    public static final String MESSAGE_QUEUE="message_queue";
    public static final String MESSAGE_EXCHANGE="message_exchange";
    public static final String MESSAGE_ROUTE_KEY="message_route_key";
     @Bean
     public Queue queue(){
         return new Queue(MESSAGE_QUEUE,true);
     }
     @Bean
     public DirectExchange directExchange(){
         return new DirectExchange(MESSAGE_EXCHANGE);
     }
     @Bean
     public Binding binding(){
         return BindingBuilder.bind(queue()).to(directExchange()).with(MESSAGE_ROUTE_KEY);
     }
     @Bean
     public MessageConverter messageConverter(){
         return new Jackson2JsonMessageConverter();
     }

    private RabbitTemplate rabbitTemplate;

   public void rabbitTemplate(){
          rabbitTemplate.setConfirmCallback(this);
          rabbitTemplate.setReturnsCallback(this);
   }

    @Primary
    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory){
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        this.rabbitTemplate = rabbitTemplate;
        rabbitTemplate.setMessageConverter(messageConverter());
        rabbitTemplate();
        return rabbitTemplate;
    }


    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String s) {
       if (ack){
            log.info("{}到达交换机",correlationData.getId());
       }else {
          log.error("{}消息丢失",correlationData.getId());
       }
    }

    @Override
    public void returnedMessage(ReturnedMessage returnedMessage) {
     log.error("{}消息未到达队列",returnedMessage.getMessage().getMessageProperties().getMessageId());
    }
}
